package cn.monkey.io.transport.mq.rabbitmq;

import cn.monkey.io.transport.core.Exchange;
import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;

public class SimpleReactiveRabbitMqExchangeFactory implements ReactiveRabbitMqExchangeFactory<RabbitMqExchange> {

    private final ReactiveRabbitMqChannelPool channelPool;

    public SimpleReactiveRabbitMqExchangeFactory(ReactiveRabbitMqChannelPool channelPool) {
        this.channelPool = channelPool;
    }

    @Override
    public Mono<RabbitMqExchange> apply(String key, Exchange exchange, RabbitMqTransportConfig rabbitMqTransportConfig) {
        return this.newInbound(rabbitMqTransportConfig.getInbound())
                .flatMap(rabbitMqInbound -> this.newOutbound(rabbitMqTransportConfig.getOutbound())
                        .map(rabbitMqOutbound -> new SimpleRabbitMqExchange(key, rabbitMqInbound, rabbitMqOutbound)));
    }

    protected Mono<RabbitMqInbound> newInbound(RabbitMqInboundConfig inbound) {
        return Flux.fromArray(inbound.getAddresses())
                .flatMap(address -> this.channelPool.findOrCreate(Address.parseAddress(address), inbound.getProperties()))
                .collectList()
                .map(channels -> new SimpleRabbitMqInbound(channels, inbound, () -> this.channelPool.destroy(channels.toArray(Channel[]::new))));
    }

    protected Mono<RabbitMqOutbound> newOutbound(RabbitMqOutboundConfig outbound) {
        return Flux.fromArray(outbound.getAddresses())
                .flatMap(address -> this.channelPool.findOrCreate(Address.parseAddress(address), outbound.getProperties()))
                .flatMap(channel -> Mono.fromCompletionStage(CompletableFuture.supplyAsync(() -> {
                    for (String topic : outbound.getTopics()) {
                        try {
                            channel.queueDeclare(topic, false, true, false, outbound.getProperties());
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    return channel;
                })))
                .collectList()
                .map(channels -> new SimpleRabbitMqOutbound(channels, outbound, () -> this.channelPool.destroy(channels.toArray(Channel[]::new))));
    }
}
